package com.hongtu.message;

import com.alibaba.fastjson.JSONObject;
import com.hongtu.dao.AccountInfoDao;
import com.hongtu.model.AccountChangeEvent;
import com.hongtu.service.AccountInfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "brank1_group")
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {

    @Autowired
    AccountInfoDao accountInfoDao;

    @Autowired
    AccountInfoService accountInfoService;

    /**
     * 消息发送成功后回调此方法，执行本地事务
     * @param message
     * @param o
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String jsonString = new String((byte[]) message.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(jsonString);

        AccountChangeEvent accountChangeEvent =
                JSONObject.parseObject(jsonObject.getString("accountChangeEvent"), AccountChangeEvent.class);
        try {
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            log.info("[rocket mq] 执行本地事务成功，commit");
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("[rocket mq] 执行本地事务失败，rollbakc");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }


    /**
     * 检查消息是否已经发送
     *
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        RocketMQLocalTransactionState state;
        Object messagePayload = message.getPayload();
        JSONObject jsonObject = JSONObject.parseObject(new String((byte[]) messagePayload));
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(jsonObject.getString("accountChangeEvent"), AccountChangeEvent.class);
        String txNo = accountChangeEvent.getTxNo();
        int existTx = accountInfoDao.isExistTx(txNo);
        log.info("[rocket mq] 检查 mq是否已经被执行，是否被执行过", txNo);
        if (existTx > 0) {
            state = RocketMQLocalTransactionState.COMMIT;
        } else {
            state = RocketMQLocalTransactionState.UNKNOWN;
        }
        return state;
    }
}
